package com.simple.mqtt.server.dispatcher;

import com.simple.mqtt.common.tool.LogTool;
import com.simple.mqtt.server.config.MqttBrokerConfig;
import com.simple.mqtt.server.domain.Message;
import com.simple.mqtt.server.domain.Subscribe;
import com.simple.mqtt.server.processor.MqttBrokerProcessor;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttQoS;

import java.util.List;
import java.util.concurrent.ExecutorService;

/**
 * MQTT 消息分发器抽象类，实现了消息分发的核心逻辑
 * 提供消息分发、发送和保留消息处理的基础实现
 */
public abstract class AbstractMqttMessageDispatcher implements IMqttMessageDispatcher {

    protected final MqttBrokerConfig config;
    protected final ExecutorService executorService;

    /**
     * 构造函数
     * @param config MQTT Broker配置
     * @param executorService 线程池，用于异步执行消息分发
     */
    public AbstractMqttMessageDispatcher(MqttBrokerConfig config, ExecutorService executorService) {
        this.config = config;
        this.executorService = executorService;
    }

    @Override
    public void dispatch(Message message, String fromClientId) {
        // 设置消息的发送者ID
        message.setClientId(fromClientId);
        // 调用send方法进行消息分发
        send(message);
    }

    @Override
    public void send(Message message) {
        try {
            String topic = message.getTopic();
            // 查找订阅该主题的客户端
            List<Subscribe> subscribeList = config.getSessionManager().searchSubscribe(topic);
            if (subscribeList.isEmpty()) {
                LogTool.debug("No subscribers for topic: {}", topic);
                return;
            }

            // 发布消息给每个订阅者
            for (Subscribe subscribe : subscribeList) {
                final String clientId = subscribe.getClientId();
                // 避免自己发给自己
                if (clientId.equals(message.getClientId())) {
                    continue;
                }

                executorService.execute(() -> {
                    try {
                        ChannelHandlerContext context = config.getChannelManager().getChannel(clientId);
                        if (context == null || !context.channel().isActive()) {
                            LogTool.warn("Client {} not connected, skip message", clientId);
                            return;
                        }

                        // 协商最终的QoS级别，取发布者和订阅者QoS的最小值
                        int subQos = subscribe.getMqttQoS();
                        MqttQoS qos = MqttQoS.valueOf(message.getQos());
                        MqttQoS finalQos = qos.value() > subQos ? MqttQoS.valueOf(subQos) : qos;

                        MqttBrokerProcessor processor = new MqttBrokerProcessor(config, config.getBroker().getScheduledExecutor());
                        processor.publishMessage(context, clientId, topic, message.getPayload(), finalQos, message.isRetain(), config.getSerializer());
                    } catch (Throwable e) {
                        LogTool.error("Error publishing message to client {}", clientId, e);
                    }
                });
            }
        } catch (Exception e) {
            LogTool.error("Error dispatching message", e);
        }
    }

    @Override
    public void sendRetainMessage(ChannelHandlerContext context, String clientId, Message message) {
        try {
            // 异步发送保留消息
            executorService.submit(() -> {
                try {
                    MqttBrokerProcessor processor = new MqttBrokerProcessor(config, config.getBroker().getScheduledExecutor());
                    processor.publishMessage(context, clientId, message.getTopic(), message.getPayload(), 
                            MqttQoS.valueOf(message.getQos()), false, config.getSerializer());
                } catch (Exception e) {
                    LogTool.error("Error sending retain message", e);
                }
            });
        } catch (Exception e) {
            LogTool.error("Error scheduling retain message", e);
        }
    }
}